package IC.kafka.consumer;

import IC.common.core.domain.R;
import IC.kafka.config.OssUtil2;
import IC.kafka.remote.*;
import com.IC.domain.po.AccessControlRecordPO;
import com.IC.domain.vo.AccessControlRecordVO;
import com.IC.domain.vo.ActualPopulationVO;
import com.alibaba.fastjson.JSONObject;
import com.ic.alarm.domain.WarnData;
import com.ic.domain.FaceDetectionResult;
import com.ic.domain.dto.FaceRecognitionDTO;
import com.ic.domain.response.FaceDetectionResponse2;
import ic.distribution.domain.Defense;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.*;

@Component
@Log4j2
public class Kafkaconsumer {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private KafkaTovideopatrol kafkaTovideopatrol;

    @Autowired
    private EsAccessRecord esAccessRecord;



    @Autowired
    private AccessControlRecord accessControlRecord;


    @Autowired
    private KafkaTodistribution kafkaTodistribution;
    @KafkaListener(topics = "data", containerFactory = "kafkaListenerContainerFactory", errorHandler = "myKafkaListenerErrorHandler",groupId = "firstGroup")
    public void kafukaconsumer(ConsumerRecord consumerRecord, Acknowledgment acknowledgment) {
        FaceRecognitionDTO faceRecognitionDTO = (FaceRecognitionDTO) consumerRecord.value();
     /*   AccessControlRecordPO accessControlRecordPO = new AccessControlRecordPO();
        accessControlRecordPO.setPic(faceRecognitionDTO.getImage());
        log.info("图片"+faceRecognitionDTO.getImage());
        accessControlRecord.add(accessControlRecordPO);*/
        esAccessRecord.indexFaceData(faceRecognitionDTO);
        acknowledgment.acknowledge();
    }



    @KafkaListener(topics = "comparisons", containerFactory = "kafkaListenerContainerFactory", errorHandler = "myKafkaListenerErrorHandler", groupId = "firstGroup2")
    public void kafukaconsumer2(ConsumerRecord consumerRecord, Acknowledgment acknowledgment) {
        try {
            WarnData warnData = new WarnData();
            FaceRecognitionDTO value = (FaceRecognitionDTO) consumerRecord.value();
            String imageurl = value.getImage();
            String image = OssUtil2.imageUrlToBase64(imageurl);
            R<FaceDetectionResponse2> faceDetectionResponse2 = kafkaTovideopatrol.imgContrast(image);
            List<FaceDetectionResult> result = faceDetectionResponse2.getData().getResult();
            R<List<Defense>> effective = kafkaTodistribution.effective();
            List<Defense> data = effective.getData();

            boolean isMatchAndAssigned = result.stream()
                    .anyMatch(faceDetectionResult -> faceDetectionResult.getSubjects().stream()
                            .anyMatch(subject -> {
                                double similarity = subject.getSimilarity();
                                if (similarity >= 0.9) {

                                    // 检查比对成功的人员是否是布控任务指定的人员
                                    Optional<Defense> matchedDefense = data.stream()
                                            .filter(defense -> defense.getDefensePersonName().equals(subject.getSubject()) && defense.getDefenseStatus() == 1 &&
                                                    isWithinTimeRange(defense.getDefenseStartDate(), defense.getDefenseEndDate()))
                                            .findFirst();

                                    if (matchedDefense.isPresent()) {
                                        warnData.setWarnPerson(matchedDefense.get().getDefensePersonName());
                                        warnData.setWarnPersonImage(matchedDefense.get().getDefensePerson());
                                        warnData.setWarnTime(new Date());
                                        warnData.setWarnType(1);
                                        warnData.setMessage("发现疑似嫌疑人");
                                        warnData.setWarnPersonName(subject.getSubject());
                                        return true;
                                    }
                                }
                                return false;
                            }));

            if (isMatchAndAssigned) {
                rabbitTemplate.convertAndSend("queue_two", JSONObject.toJSONString(warnData), message -> {
                    message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
                    return message;
                });
            }
            acknowledgment.acknowledge();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 消费门禁记录，将门禁记录存入es
     * @param consumerRecord
     * @param acknowledgment
     */
    @KafkaListener(topics = "accessControlRecord", containerFactory = "kafkaListenerContainerFactory", errorHandler = "myKafkaListenerErrorHandler",groupId = "firstGroup3")
    public void kafukaconsumer3(ConsumerRecord consumerRecord, Acknowledgment acknowledgment) {
        List<AccessControlRecordVO> value = (List<AccessControlRecordVO>) consumerRecord.value();
        esAccessRecord.indexAccessRecords(value);
        acknowledgment.acknowledge();
    }

    /**
     * 消费人口数据，将人口数据存入es
     * @return
     */
    @KafkaListener(topics = "Population", containerFactory = "kafkaListenerContainerFactory", errorHandler = "myKafkaListenerErrorHandler",groupId = "firstGroup4")
    public void kafukaconsumer4(ConsumerRecord consumerRecord, Acknowledgment acknowledgment) {
        List<Map> value = (List<Map>) consumerRecord.value();
        esAccessRecord.indexList(value);
        acknowledgment.acknowledge();
    }

    /**
     * 消费住宅数据，将住宅数据存入es
     * @param consumerRecord
     * @param acknowledgment
     */
    @KafkaListener(topics = "Tenement", containerFactory = "kafkaListenerContainerFactory", errorHandler = "myKafkaListenerErrorHandler",groupId = "firstGroup5")
    public void kafukaconsumer5(ConsumerRecord consumerRecord, Acknowledgment acknowledgment) {
        List<Map> value = (List<Map>) consumerRecord.value();
        esAccessRecord.indexList(value);
        acknowledgment.acknowledge();
    }

    /**
     * 消费车辆数据，将车辆数据存入es
     * @param consumerRecord
     * @param acknowledgment
     */
    @KafkaListener(topics = "vehicle", containerFactory = "kafkaListenerContainerFactory", errorHandler = "myKafkaListenerErrorHandler",groupId = "firstGroup6")
    public void kafukaconsumer6(ConsumerRecord consumerRecord, Acknowledgment acknowledgment) {
        List<Map> value = (List<Map>) consumerRecord.value();
        esAccessRecord.indexList(value);
        acknowledgment.acknowledge();
    }


    private boolean isWithinTimeRange(LocalDateTime startTime, LocalDateTime endTime) {
        LocalDateTime now = LocalDateTime.now();
        return !now.isBefore(startTime) && !now.isAfter(endTime);
    }
}
